package com.ytktt.study.week12.task6;

import com.ytktt.study.week12.task6.exception.KmqException;

import java.util.concurrent.TimeUnit;

/**
 * Kmq核心类
 *
 * @author ytk
 */
public final class Kmq<T> {

    public Kmq(String topic, int capacity) {
        this.topic = topic;
        this.capacity = capacity;
        this.buffer = new KmqBuffer<>(capacity);
    }

    private final String topic;

    private final int capacity;

    private final KmqBuffer<KmqMessage<T>> buffer;

    public boolean send(KmqMessage<T> message) {
        return this.buffer.offer(message);
    }

    public KmqMessage<T> poll(String name) {
        return this.buffer.poll(name);
    }

    public KmqMessage<T> poll(String name, long timeout, TimeUnit timeUnit) {
        try {
            return this.buffer.poll(name, timeout, timeUnit);
        } catch (InterruptedException e) {
            throw new KmqException("consumer poll message error", e);
        }
    }

    public void registerConsumer(String name) {
        this.buffer.registerConsumer(name);
    }

    public void seek(String name, int offset) {
        this.buffer.seek(name, offset);
    }
}
